package com.sunday.common.reactor.e9_Splitting_Flux;

import com.sunday.common.reactor.SubscribeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.GroupedFlux;

import java.time.Duration;
import java.util.ArrayList;

/**
 * A.7. 拆分通量
 */
@Slf4j
public class SplittingFlux {

    /**
     * 我想按边界条件将 Flux<T> 拆分为 ：Flux<Flux<T>>
     * 大小：window（int）
     * ...​具有重叠或删除窗口：window（int， int）
     * 时间窗口（持续时间）
     * ...​具有重叠或拖放窗口：window（Duration， Duration）
     * 大小或时间（达到计数或超时时关闭窗口）：windowTimeout（int， Duration）
     * 基于元素的谓词：windowUntil
     * ...​...在下一个窗口中发出触发边界的元素 （ variant）： .windowUntil（predicate， true）cutBefore
     * ...​当元素与谓词匹配时保持窗口打开：windowWhile（不发出不匹配的元素）
     * 由控件 Publisher 中 onNexts 表示的任意边界驱动：window（Publisher）、windowWhen
     */


    /**
     * 我想按边界条件将 Flux<T> 拆分为 ：Flux<Flux<T>>
     * 大小：window（int）
     * ...​具有重叠或删除窗口：window（int， int）
     */
    @Test
    public void window() {

        log.info("我想按边界条件将 Flux<T> 拆分为 ：Flux<Flux<T>>");
        // 我想按边界条件将 Flux<T> 拆分为 ：Flux<Flux<T>>
        Flux<Flux<Integer>> flux = Flux.range(1, 100)
                // 按照每20个元素拆分
                .window(20);

        flux.subscribe(innerFlux -> {
            log.info("innerFlux: {}", innerFlux);
            innerFlux.subscribe(i -> log.info("innerFlux: {}", i));
        });

    }

    /**
     * 时间窗口（持续时间）
     * ...​具有重叠或拖放窗口：window（Duration， Duration）
     */
    @Test
    public void window2() throws InterruptedException {

        log.info("时间窗口（持续时间）");
        // 时间窗口（持续时间）
        Flux<Flux<Long>> flux = Flux.interval(Duration.ofMillis(100))
                // 每隔100毫秒拆分
                .window(Duration.ofMillis(500));

        flux.subscribe(innerFlux -> {
            log.info("innerFlux: {}", innerFlux);
            innerFlux.subscribe(i -> log.info("innerFlux: {}", i));
        });

        Thread.sleep(5000);

    }

    /**
     * 大小或时间（达到计数或超时时关闭窗口）：windowTimeout（int， Duration）
     */
    @Test
    public void windowTimeout() throws InterruptedException {

        log.info("大大小或时间（达到计数或超时时关闭窗口）：windowTimeout（int， Duration）");
        // 大小或时间（达到计数或超时时关闭窗口）：windowTimeout（int， Duration）
        Flux<Flux<Integer>> flux = Flux.range(1, 100)
                // 每隔20个元素或者超时500毫秒拆分
                .windowTimeout(20, Duration.ofMillis(500));

        flux.subscribe(innerFlux -> {
            log.info("innerFlux: {}", innerFlux);
            innerFlux.subscribe(i -> log.info("innerFlux: {}", i));
        });

        Thread.sleep(5000);

    }

    /**
     * 基于元素的谓词：windowUntil
     * ...​...在下一个窗口中发出触发边界的元素 （ variant）： .windowUntil（predicate， true）cutBefore
     * cutBefore 与 windowUntil 的区别在于，它在窗口中发出触发边界的元素，而不是在下一个窗口中发出它们。 `
     * ...​当元素与谓词匹配时保持窗口打开：windowWhile（不发出不匹配的元素）
     */
    @Test
    public void windowUntil() {

//        log.info("基于元素的谓词：windowUntil");
//        // 基于元素的谓词：windowUntil
//        // ...​...在下一个窗口中发出触发边界的元素 （ variant）： .windowUntil（predicate， true）cutBefore
//        Flux<Flux<Integer>> flux = Flux.range(1, 100)
//                // 每隔20个元素
//                .windowUntil(i -> i % 20 == 0);
//
//        flux.subscribe(innerFlux -> {
//            log.info("innerFlux: {}", innerFlux);
//            innerFlux.subscribe(i -> log.info("innerFlux: {}", i));
//        });

        log.info("当元素与谓词匹配时保持窗口打开：windowWhile（不发出不匹配的元素）");
        // ...​当元素与谓词匹配时保持窗口打开：windowWhile（不发出不匹配的元素）
        Flux<Flux<Integer>> flux2 = Flux.range(1, 100)
                // windowWhile 与 windowUntil 的区别在于，它在窗口中发出触发边界的元素，而不是在下一个窗口中发出它们。
                .windowWhile(i -> i % 20 == 0);

        flux2.subscribe(innerFlux -> {
//            log.info("innerFlux: {}", innerFlux);
            innerFlux.subscribe(i -> log.info("innerFlux: {}", i));
        });

    }

    /**
     * 我想将 Flux<T 拆分>并将边界内的元素缓冲在一起......
     * 进入列表：
     * 按大小边界：buffer（int）
     * ...​使用重叠或删除缓冲区：buffer（int， int）
     * 按持续时间边界：buffer（Duration）
     * ...​使用重叠或删除缓冲区：buffer（Duration， Duration）
     * 按大小或持续时间边界：bufferTimeout（int， Duration）
     * 按任意条件边界：bufferUntil（Predicate）
     * ...​将触发边界的元素放在下一个缓冲区中：.bufferUntil（predicate， true）
     * ...​谓词匹配时缓冲并删除触发边界的元素：bufferWhile（Predicate）
     * 由控件发布服务器中 onNexts 表示的任意边界驱动：buffer（Publisher）、bufferWhen
     * 到任意“集合”类型：使用像 buffer（int， Supplier<C>） 这样的变体C
     */
    @Test
    public void buffer() {
        log.info("我想将 Flux<T 拆分>并将边界内的元素缓冲在一起......");
        // 我想将 Flux<T 拆分>并将边界内的元素缓冲在一起......
        // 进入列表：
        // 按大小边界：buffer（int）
        // ...​使用重叠或删除缓冲区：buffer（int， int）
        log.info("按大小边界：buffer（int）");
        Flux<Integer> flux = Flux.range(1, 100);
        // 每隔20个元素
        SubscribeUtils.subscribe(flux.buffer(20));

        // 使用重叠或删除缓冲区：buffer（int， int）
        log.info("使用重叠或删除缓冲区：buffer（int， int）");
        // 每隔20个元素，重叠10个元素
        SubscribeUtils.subscribe(flux.buffer(20, 10));
    }

    /**
     * 按持续时间边界：buffer（Duration）
     * ...​使用重叠或删除缓冲区：buffer（Duration， Duration）
     */
    @Test
    public void bufferDuration() throws InterruptedException {
        log.info("按持续时间边界：buffer（Duration）");
        // 按持续时间边界：buffer（Duration）
        Flux<Long> flux = Flux.interval(Duration.ofMillis(100));
        // 每隔500毫秒
        SubscribeUtils.subscribe(flux.buffer(Duration.ofMillis(500)));

        Thread.sleep(5000);
    }

    @Test
    public void bufferOverlapDuration() throws InterruptedException {
        log.info("使用重叠或删除缓冲区：buffer（Duration， Duration）");
        // 使用重叠或删除缓冲区：buffer（Duration， Duration）
        Flux<Long> flux = Flux.interval(Duration.ofMillis(100));
        // 每隔500毫秒，重叠200毫秒
        SubscribeUtils.subscribe(flux.buffer(Duration.ofMillis(500), Duration.ofMillis(200)));

        Thread.sleep(5000);
    }

    /**
     * 按大小或持续时间边界：bufferTimeout（int， Duration）
     */
    @Test
    public void bufferTimeout() throws InterruptedException {
        log.info("按大小或持续时间边界：bufferTimeout（int， Duration）");
        // 按大小或持续时间边界：bufferTimeout（int， Duration）
        Flux<Long> flux = Flux.interval(Duration.ofMillis(100));
        // 每隔20个元素或者超时500毫秒
        SubscribeUtils.subscribe(flux.bufferTimeout(20, Duration.ofMillis(500)));

        Thread.sleep(5000);
    }

    /**
     * 按任意条件边界：bufferUntil（Predicate）
     * ...​将触发边界的元素放在下一个缓冲区中：.bufferUntil（predicate， true）
     * ...​谓词匹配时缓冲并删除触发边界的元素：bufferWhile（Predicate）
     */
    @Test
    public void bufferUntil() {
        log.info("按任意条件边界：bufferUntil（Predicate）");
        // 按任意条件边界：bufferUntil（Predicate）
        // ...​将触发边界的元素放在下一个缓冲区中：.bufferUntil（predicate， true）
        // ...​谓词匹配时缓冲并删除触发边界的元素：bufferWhile（Predicate）
        Flux<Integer> flux = Flux.range(1, 100);
        // 每隔20个元素
        SubscribeUtils.subscribe(flux.bufferUntil(i -> i % 20 == 0));
        // Predicate<? super T> predicate, boolean cutBefore cutBefore 与 bufferUntil 的区别在于，它在窗口中发出触发边界的元素，而不是在下一个窗口中发出它们。
        // cutBefore 边界是在那个窗口
        SubscribeUtils.subscribe(flux.bufferUntil(i -> i % 20 == 0, true));
    }

    @Test
    public void bufferWhile() {
        log.info("谓词匹配时缓冲并删除触发边界的元素：bufferWhile（Predicate）");
        // 谓词匹配时缓冲并删除触发边界的元素：bufferWhile（Predicate）
        Flux<Integer> flux = Flux.range(1, 100);
        // 每隔20个元素
        SubscribeUtils.subscribe(flux.bufferWhile(i -> i % 20 == 0));
    }

    /**
     * 到任意“集合”类型：使用像 buffer（int， Supplier<C>） 这样的变体C
     */
    @Test
    public void bufferSupplier() {
        log.info("到任意“集合”类型：使用像 buffer（int， Supplier<C>） 这样的变体C");
        // 到任意“集合”类型：使用像 buffer（int， Supplier<C>） 这样的变体C
        Flux<Integer> flux = Flux.range(1, 100);
        // 每隔20个元素
        SubscribeUtils.subscribe(flux.buffer(20, ArrayList::new));
    }

    /**
     * 我想拆分一个 Flux<T>以便共享特征的元素最终出现在同一个子 flux 中： groupBy（Function<T，K>）
     * 提示：请注意，这将返回一个 ，每个内部 GroupedFlux 共享可通过 key（） 访问的相同键。Flux<GroupedFlux<K, T>>K
     */
    @Test
    public void groupBy() {
        log.info("我想拆分一个 Flux<T>以便共享特征的元素最终出现在同一个子 flux 中： groupBy（Function<T，K>） 提示：请注意，这将返回一个 ，每个内部 GroupedFlux 共享可通过 key（） 访问的相同键。Flux<GroupedFlux<K, T>>K");
        // 我想拆分一个 Flux<T>以便共享特征的元素最终出现在同一个子 flux 中： groupBy（Function<T，K>） 提示：请注意，这将返回一个 ，每个内部 GroupedFlux 共享可通过 key（） 访问的相同键。Flux<GroupedFlux<K, T>>K
        Flux<GroupedFlux<Integer, Integer>> flux = Flux.range(1, 100)
                // 每隔20个元素
                .groupBy(i -> i % 20);

        flux.subscribe(groupedFlux -> {
            log.info("groupedFlux.key(): {}", groupedFlux.key());
            groupedFlux.subscribe(i -> log.info("groupedFlux.key() : {} , groupedFlux: {}",groupedFlux.key(), i));
        });
    }

}
